package edu.hit.pact.pgse.master.handler;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IoSession;



import edu.hit.pact.data.SpeedData;
import edu.hit.pact.pgse.bean.TaskPieceStatistics;
import edu.hit.pact.pgse.communication.message.SpeedReportMessage;
import edu.hit.pact.pgse.communication.message.TaskPieceStatisticsMessage;
import edu.hit.pact.pgse.master.MasterManager;
import edu.hit.pact.pgse.master.util.CrawlerDeadHandler;
import edu.hit.pact.pgse.util.Globals;
import edu.hit.pact.plot.PlotProcessor;

public class MasterCrawlerHandler extends IoHandlerAdapter{
	private Log logger;
	private MasterManager masterManager;
	
	public static ConcurrentHashMap<String, Integer> latestSpeeds = new ConcurrentHashMap<String, Integer>();

	public MasterCrawlerHandler(MasterManager masterManager){
		this.logger = LogFactory.getLog(MasterCrawlerHandler.class);
		this.masterManager = masterManager;
	}
	
	public void sessionOpened(IoSession session) {
		logger.error("session is opened, master connected to the crawler!"+ session);
		masterManager.processCrawlerRegister(session);
		masterManager.operate("crawler", "add", session);
	}//end for
	
	public void sessionClosed(IoSession session) throws Exception {
    	logger.error("sessionClosed, master can not connect to the crawler!"+ session);
		masterManager.operate("crawler", "remove", session);
		String crawlerAddress = session.getRemoteAddress().toString();
		
		if (latestSpeeds.containsKey(crawlerAddress)){
			latestSpeeds.put("Master",latestSpeeds.get("Master") - latestSpeeds.get(crawlerAddress));
			masterManager.removePlotNode(session.getRemoteAddress().toString());
			latestSpeeds.remove(crawlerAddress);
		}
		
		masterManager.processCrawlerDead(session);
		InetAddress ip = ((InetSocketAddress)session.getRemoteAddress()).getAddress();
    	session.close(true);
    	CrawlerDeadHandler crawlerDeadHandler = new CrawlerDeadHandler(masterManager, ip);
    	crawlerDeadHandler.start();
	}
	
	public void messageReceived(IoSession session, Object message) {
    	if (message instanceof TaskPieceStatisticsMessage){
    		logger.info("masterManager processFinishedNotify");
        	TaskPieceStatistics statistics = ((TaskPieceStatisticsMessage)message).getStatistics();
        	masterManager.processFinishedNotify(statistics);
    	} else if (message instanceof SpeedReportMessage){
    		int speed = ((SpeedReportMessage)message).getCurrentSpeed();
    		String crawlerAddress = session.getRemoteAddress().toString();

    		if (latestSpeeds.containsKey("Master")){
    			int currentMasterSpeed = 0;
    			
    			if (latestSpeeds.containsKey(crawlerAddress)){
    				currentMasterSpeed = latestSpeeds.get("Master") - latestSpeeds.get(crawlerAddress)+speed;
    			} else {
    				currentMasterSpeed = latestSpeeds.get("Master")+ speed;
    			}
    			
    			latestSpeeds.put("Master", currentMasterSpeed);
    		} else {//if master doesn't exist, should insert a new master record, and set master's speed equals to speed
    			
    			latestSpeeds.put("Master", speed);
    		}
    		
    		latestSpeeds.put(crawlerAddress, speed);
    		
    		masterManager.processSpeedReport(crawlerAddress, Globals.MASTER_ADDRESS_IP, speed);
    	} else {
    		logger.error("master recived a error message from crawler!"+ session);
    	}
    	
    }
	
    public void exceptionCaught(IoSession session, Throwable cause) {
    	logger.error("exceptionCaught, master can not connect to the crawler! "+ session);
    	cause.printStackTrace();
    	logger.error(cause.getStackTrace());
    	session.close(true);
//    	MasterManager.sessionToCrawlers.put(session.getRemoteAddress(), null);
    }
}
